package top.wintp.offlinedataanalysis.anlyser.mr.nu;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.ibatis.session.SqlSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import top.wintp.offlinedataanalysis.anlyser.dao.StatsDeviceBrowserDAO;
import top.wintp.offlinedataanalysis.anlyser.dao.StatsUserDAO;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsCommonDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.BrowserDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.DateDimension;
import top.wintp.offlinedataanalysis.anlyser.dim.base.PlatformDimension;
import top.wintp.offlinedataanalysis.anlyser.domain.StatsDeviceBrowser;
import top.wintp.offlinedataanalysis.anlyser.domain.StatsUser;
import top.wintp.offlinedataanalysis.anlyser.value.reduce.MapWritableValue;
import top.wintp.offlinedataanalysis.common.KpiType;

/**
 * @author: pyfysf
 * <p>
 * @qq: 337081267
 * <p>
 * @CSDN: http://blog.csdn.net/pyfysf
 * <p>
 * @blog: http://wintp.top
 * <p>
 * @email: pyfysf@163.com
 * <p>
 * @time: 2019/8/27
 */
public class MysqlRecordWriter extends RecordWriter<StatsUserDimension, MapWritableValue> {
    private static final Logger logger = LoggerFactory.getLogger(MysqlRecordWriter.class);
    private final SqlSession session;

    private final MySqlCollector collector;

    /**
     * 缓存
     */
    private List<StatsUser> recordNewInstallUserCacheList = new ArrayList<>();
    private List<StatsDeviceBrowser> browserNewInstallUserCacheList = new ArrayList<>();
    private Writable mapValueKey = new IntWritable(-1);

    public MysqlRecordWriter(MySqlCollector collector) {
        this.collector = collector;
        this.session = collector.getSession();
    }

    @Override
    public void write(StatsUserDimension key, MapWritableValue value) throws IOException, InterruptedException {
        String kpiName = value.getKpi().name;
        //浏览器的数据
        BrowserDimension browser = key.getBrowser();
        Integer browserId = this.collector.convertBrowserData(browser);

        StatsCommonDimension statsCommon = key.getStatsCommon();
        //日期数据
        DateDimension date = statsCommon.getDate();
        Integer dateId = this.collector.convertDateData(date);
        //platform
        PlatformDimension platformDimension = statsCommon.getPlatform();
        Integer platformId = this.collector.convertPlatformData(platformDimension);
        //数据
        IntWritable newInstallUserCount = (IntWritable) value.getValue().get(mapValueKey);
        //保存用户信息
        StatsUser record = new StatsUser();
        record.setDateDimensionId(dateId);
        record.setPlatformDimensionId(platformId);
        record.setNewInstallUsers(newInstallUserCount.get());

        StatsDeviceBrowser statsDeviceBrowser = new StatsDeviceBrowser();
        statsDeviceBrowser.setPlatformDimensionId(platformId);
        statsDeviceBrowser.setDateDimensionId(dateId);
        statsDeviceBrowser.setBrowserDimensionId(browserId);
        statsDeviceBrowser.setNewInstallUsers(newInstallUserCount.get());

        if (KpiType.NEW_INSTALL_USER.name.equalsIgnoreCase(kpiName)) {
//            新增用户
            recordNewInstallUserCacheList.add(record);
        } else if (KpiType.BROWSER_NEW_INSTALL_USER.name.equals(kpiName)) {
            browserNewInstallUserCacheList.add(statsDeviceBrowser);
        }

//        批量插入
        if (recordNewInstallUserCacheList.size() > 0 && recordNewInstallUserCacheList.size() % 10 == 0) {
            insertRecordNewInstallUser();
        }

//        批量插入
        if (browserNewInstallUserCacheList.size() > 0 && browserNewInstallUserCacheList.size() % 10 == 0) {
            insertBrowserNewInstallUser();
        }


    }

    /**
     * 插入浏览器用户
     */
    private void insertBrowserNewInstallUser() {
        StatsDeviceBrowserDAO statsDeviceBrowserDAO = this.session.getMapper(StatsDeviceBrowserDAO.class);

        logger.info("MysqlRecordWriter  write()  插入浏览器用户 start   ");
//            10条数据插入一次
        statsDeviceBrowserDAO.batchInsertDataSelective(browserNewInstallUserCacheList);
        this.session.commit();
        logger.info("MysqlRecordWriter  write()   插入浏览器用户 stop   ");
        browserNewInstallUserCacheList.clear();
    }

    /**
     * 插入新增用户
     */
    private void insertRecordNewInstallUser() {
        StatsUserDAO statsUserDAO = this.session.getMapper(StatsUserDAO.class);

        logger.info("MysqlRecordWriter  write()   插入新增用户 start   ");
//            10条数据插入一次
        statsUserDAO.batchInsertDataSelective(recordNewInstallUserCacheList);
        this.session.commit();
        logger.info("MysqlRecordWriter  write()   插入新增用户 stop   ");
        recordNewInstallUserCacheList.clear();
    }


    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
//        将没有写出的数据继续写出
        if (recordNewInstallUserCacheList.size() > 0) {
            insertRecordNewInstallUser();
        }

        if (browserNewInstallUserCacheList.size() > 0) {
            insertBrowserNewInstallUser();
        }

        logger.info("MysqlRecordWriter  close()   关闭释放资源时 插入残余数据   ");


//        关闭资源
        if (this.session != null) {
            this.session.close();
        }

    }
}
